package com.yuebook.smartrpc.framework.provider;

import com.google.common.collect.Maps;
import com.yuebook.smartrpc.framework.model.ProviderService;
import com.yuebook.smartrpc.framework.model.SmartRpcRequest;
import com.yuebook.smartrpc.framework.model.SmartRpcResponse;
import com.yuebook.smartrpc.framework.zookeeper.IRegisterCenterProvider;
import com.yuebook.smartrpc.framework.zookeeper.RegisterCenter;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import java.lang.reflect.Method;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

/**
 * @author honglei
 * @date 2018/11/19 下午9:33
 */
public class NettyServerInvokeHandler extends SimpleChannelInboundHandler<SmartRpcRequest> {

    //服务端限流
    private static Map<String, Semaphore> serviceKeySemaphoreMap = Maps.newConcurrentMap();

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.close();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, SmartRpcRequest request)
        throws Exception {
        if (ctx.channel().isWritable()) {
            ProviderService providerService = request.getProviderService();
            String methodName = request.getInvokedMethodName();
            long invokeTimeout = request.getInvokeTimeout();

            //获取限流
            String serviceKey = providerService.getServiceItf().getName();

            IRegisterCenterProvider registerCenter = RegisterCenter.singleton();
            List<ProviderService> providerServiceList = registerCenter.getProviderServiceMap().get(serviceKey);
            ProviderService localProviderService = null;
            for (ProviderService p : providerServiceList) {
                if (p.getServiceMethod().getName().equals(methodName)) {
                    localProviderService = p;
                    break;
                }
            }

            Semaphore semaphore = serviceKeySemaphoreMap.get(serviceKey);
            if (semaphore == null) {
                synchronized (serviceKeySemaphoreMap) {
                    semaphore = serviceKeySemaphoreMap.get(serviceKey);
                    if (semaphore == null) {
                        semaphore = new Semaphore(localProviderService.getWorkerThreads());
                        serviceKeySemaphoreMap.put(serviceKey, semaphore);
                    }
                }
            }

            Object serviceObj = localProviderService.getServiceObject();
            Object[] args = request.getArgs();
            Method method = localProviderService.getServiceMethod();
            Object result = null;
            boolean acquire = false;
            try {
                acquire = semaphore.tryAcquire(invokeTimeout, TimeUnit.MILLISECONDS);
                if (acquire) {
                    result = method.invoke(serviceObj, args);
                }
            } catch (Exception e) {
                result = e;
            } finally {
                if (acquire) {
                    semaphore.release();
                }
            }
            SmartRpcResponse response = new SmartRpcResponse();
            response.setUniqueKey(request.getUniqueKey());
            response.setInvokeTimeout(invokeTimeout);
            response.setResult(result);

            ctx.channel().writeAndFlush(response);
        }
    }
}
